package com.test.spring.sliding;

import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 滑动窗口
 *
 * @Author ll
 * @Date 2017/7/11 14:31
 */
public final class SlidingWindowCounter<T> implements Serializable {

    private static final long serialVersionUID = 1430908048226167541L;

    private SlotBasedCounter<T> objCounter;

    private volatile int headSlot;

    private volatile int tailSlot;

    private volatile int windowSize;

    private ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor();

    public SlidingWindowCounter(final int windowSize, long period, TimeUnit unit) {
        resizeWindow(windowSize);
        ses.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                resizeWindow(windowSize);
            }
        }, period, period, unit);
    }

    private synchronized void resizeWindow(int windowSize) {
        this.windowSize = windowSize < 2 ? 2 : windowSize;
        this.objCounter = new SlotBasedCounter<>(this.windowSize);
        this.headSlot = 0;
        this.tailSlot = slotAfter(headSlot);
    }

    public void incrementCount(T t) {
        objCounter.incrementCount(t, headSlot);
    }

    public long getTotalCount(T t) {
        return objCounter.getTotalCount(t);
    }

    public Map<T, Long> getCountsThenAdvanceWindow() {
        Map<T, Long> counts = objCounter.getCounts();
        objCounter.wipeZeros();
        objCounter.wipeSlot(tailSlot);
        advanceHead();
        return counts;
    }

    private void advanceHead() {
        headSlot = tailSlot;
        tailSlot = slotAfter(tailSlot);
    }

    private int slotAfter(int slot) {
        return (slot + 1) % windowSize;
    }
}
